package com.spx.chapter05.source;

import com.spx.chapter05.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;

/**
 * create by undeRdoG on  2022-04-29  15:14
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class CustomSourceFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Event> source = env.addSource(new MySourceFunction()).setParallelism(1);

        source.print("custom SourceFunction");

        env.execute();
    }




    /*
    *  此处的SourceFunction 实现了SourceFunction 接口，并行度只能为1，如果调大了并行度，会抛出以下异常
    *  Exception in thread "main" java.lang.IllegalArgumentException: The parallelism of non parallel operator must be 1.
    *  如要提高source的并行度，自定义的Source需要实现 ParallelSourceFunction
    * */

    public static class MySourceFunction implements SourceFunction<Event> {

        // 默认为运行状态
        boolean isActive = true;
        // 生成10条数据
        int count = 10;
        // 初始化条数
        int start = 0;

        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            //定义随机生成器
            Random random = new Random();
            String[] users = new String[]{"Alice","Bob","Mary","Cary"};
            String[] urls = new String[]{"./home","./cart","./fav","./prod?id=100"};

            while (isActive){
                sourceContext.collect(new Event(
                        users[random.nextInt(users.length)],
                        urls[random.nextInt(urls.length)],
                        Calendar.getInstance().getTimeInMillis()
                        ));
                start++;
                // if count equals start，cancel running
                if (start == count){
                    cancel();
                }

                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            isActive = false;
        }
    }
}
